package com.huan.kafka.config;

import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.Map;

@Configuration
public class KafkaConfig {

    // 1. 注入 Spring Boot 自动配置的 KafkaProperties
    private final KafkaProperties kafkaProperties;

    public KafkaConfig(KafkaProperties kafkaProperties) {
        this.kafkaProperties = kafkaProperties;
    }

    // 2. 基于 KafkaProperties 创建 ConsumerFactory
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        // Spring Boot 会自动将 application.properties 中的配置绑定到 kafkaProperties 对象上
        return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties());
    }

    // 3. kafkaListenerContainerFactory 依赖 consumerFactory
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {

        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();

        // 设置 consumerFactory
        factory.setConsumerFactory(consumerFactory);

        // 开启批量监听
        factory.setBatchListener(true);

        // 设置并发数, 每个分区一个线程去消费
        factory.setConcurrency(10);

        // 设置手动 Ack 模式
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

        return factory;
    }
}